package com.swak.reactivex.transport;

import java.io.IOException;
import java.util.List;

import org.springframework.util.Assert;

import com.swak.reactivex.transport.exception.CodecException;
import com.swak.registry.URL;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.DefaultFileRegion;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import lombok.Getter;

/**
 * Netty Codec.
 * 
 * @author lifeng
 */
public final class NettyCodec implements Codec {

	@Getter
	private final ChannelHandler encoder = new InternalEncoder();
	@Getter
	private final ChannelHandler decoder = new InternalDecoder();
	@Getter
	private final URL url;
	@Getter
	private final com.swak.reactivex.transport.ChannelHandler handler;

	private final Codec codec;

	public NettyCodec(Codec codec, URL url, com.swak.reactivex.transport.ChannelHandler handler) {
		Assert.isTrue(codec != null, "codec == null");
		Assert.isTrue(url != null, "url == null");
		Assert.isTrue(handler != null, "handler == null");
		this.codec = codec;
		this.url = url;
		this.handler = handler;
	}

	/**
	 * 编码 ：写流程会触发编码，但是编码阶段发生的异常只能通过 
	 * 
	 * 处理方式1： 等待 channel.writeAndFlush(message) 结束后才能知道是否有异常（最好的方式是在这里处理）
	 * 处理方式2： 也可以通过 fireExceptionCaught 让异常通过 exceptionCaught(属于入站流程) 来处理；
	 * 
	 * @param channel
	 * @param buffer
	 * @param message
	 * @throws IOException
	 */
	public void encode(Channel channel, ByteBuf buffer, Object message) throws CodecException {
		this.codec.encode(channel, buffer, message);
	}

	/**
	 * 解码：解码属于 入站流程中的一环，所以期间产生的异常会触发 exceptionCaught
	 * 
	 * @param channel
	 * @param buffer
	 * @return
	 * @throws IOException
	 */
	public Object decode(Channel channel, ByteBuf buffer) throws CodecException {
		return this.codec.decode(channel, buffer);
	}

	/**
	 * 编码
	 * 
	 * @author lifeng
	 */
	class InternalEncoder extends MessageToByteEncoder<Object> {

		@Override
		public boolean acceptOutboundMessage(Object msg) throws Exception {
			return !(msg instanceof DefaultFileRegion) && super.acceptOutboundMessage(msg);
		}

		@Override
		protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf buffer) throws Exception {
			NettyChannel channel = NettyChannel.getOrAddChannel(url, ctx.channel(), handler);
			NettyCodec.this.encode(channel, buffer, msg);
		}
	}

	/**
	 * 解码
	 * 
	 * @author lifeng
	 */
	class InternalDecoder extends ByteToMessageDecoder {

		@Override
		protected void decode(ChannelHandlerContext ctx, ByteBuf message, List<Object> out) throws Exception {
			NettyChannel channel = NettyChannel.getOrAddChannel(url, ctx.channel(), handler);
			// decode object.
			do {
				int saveReaderIndex = message.readerIndex();
				Object msg = NettyCodec.this.decode(channel, message);
				if (msg == DecodeResult.NEED_MORE_INPUT) {
					message.readerIndex(saveReaderIndex);
					break;
				} else if (msg == DecodeResult.SKIP_SOME_INPUT) {
					break;
				} else {
					// is it possible to go here ?
					if (saveReaderIndex == message.readerIndex()) {
						throw new IOException("Decode without read data.");
					}
					if (msg != null) {
						out.add(msg);
					}
				}
			} while (message.isReadable());
		}
	}
}
